﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;

namespace RabbitMQ.Extension
{
    public class RabbitConsumer
    {
        private RabbitType rabbitType;
        private IModel channel;
        private readonly object locker = new object();
        private string queue;

        public RabbitConsumer(RabbitType rabbitType, IModel channel)
        {
            this.rabbitType = rabbitType;
            this.channel = channel;
        }

        public IModel Channel
        {
            get
            {
                return channel;
            }
        }

        public void Bind(string queue = null, string exchange = null, List<string> routingKeyList = null, bool durable = false, bool autoDelete = false, int ttlSecond = 0, string deadLetterExchange = null, string deadLetterRoutingKey = null)
        {
            try
            {
                this.queue = queue;
                Dictionary<string, object> dict = null;
                if (ttlSecond > 0)
                {
                    dict = new Dictionary<string, object>
                    {
                        { "x-message-ttl",ttlSecond*1000}
                    };
                    if (!string.IsNullOrEmpty(deadLetterExchange))
                    {
                        dict.Add("x-dead-letter-exchange", deadLetterExchange);
                    }
                    if (!string.IsNullOrEmpty(deadLetterRoutingKey))
                    {
                        dict.Add("x-dead-letter-routing-key", deadLetterRoutingKey);
                    }
                }
                if (rabbitType == RabbitType.Work)
                {
                    if (string.IsNullOrEmpty(queue))
                    {
                        throw new Exception("queue can not be null!");
                    }
                    channel.QueueDeclare(queue: queue,
                        durable: durable,
                        exclusive: false,
                        autoDelete: autoDelete,
                        arguments: dict);
                    return;
                }

                if (string.IsNullOrEmpty(exchange))
                {
                    throw new Exception("exchange can not be null!");
                }

                if (string.IsNullOrEmpty(queue))
                {
                    this.queue = channel.QueueDeclare(durable: durable, arguments: dict).QueueName;
                }
                else
                {
                    channel.QueueDeclare(queue: this.queue,
                                     durable: durable,
                                     exclusive: false,
                                     autoDelete: autoDelete,
                                     arguments: dict);
                }

                if (rabbitType == RabbitType.SubPub)
                {
                    channel.ExchangeDeclare(exchange, "fanout", durable, autoDelete);
                    channel.QueueBind(queue: queue, exchange: exchange, routingKey: "");
                    return;
                }

                if (routingKeyList == null || routingKeyList.Count == 0)
                {
                    throw new Exception("routingKeyList can not be null or empty");
                }

                if (rabbitType == RabbitType.Routing)
                {
                    channel.ExchangeDeclare(exchange, "direct", durable, autoDelete);
                }
                else
                {
                    channel.ExchangeDeclare(exchange, "topic", durable, autoDelete);
                }

                foreach (var routingKey in routingKeyList)
                {
                    channel.QueueBind(queue: this.queue, exchange: exchange, routingKey: routingKey);
                }
            }
            catch (Exception ex)
            {
                channel.Dispose();
                throw ex;
            }
        }

        public bool BasicAck(ulong messageId, bool multiple = false)
        {
            lock (locker)
            {
                try
                {
                    //multiple 为true的时候表示之前messageId也都提交
                    channel.BasicAck(messageId, multiple);
                    return true;
                }
                catch
                {
                    return false;
                }
            }
        }

        public bool BasicReject(ulong messageId, bool requeue = true)
        {
            lock (locker)
            {
                try
                {
                    //requeue false扔掉消息
                    //requeue true退回消息(无限循环)
                    channel.BasicReject(messageId, requeue);
                    return true;
                }
                catch
                {
                    return false;
                }

            }
        }

        public bool BasicRecover(bool requeue = false)
        {
            lock (locker)
            {
                try
                {
                    //补发消息 true退回到queue中/false只补发给当前的consumer
                    channel.BasicRecover(requeue);
                    return true;
                }
                catch
                {
                    return false;
                }
            }
        }

        public uint GetCount()
        {
            lock (locker)
            {
                try
                {
                    var response = channel.QueueDeclarePassive(queue);
                    return response.MessageCount;
                }
                catch
                {
                    return 0;
                }
            }
        }

        public bool DeleteAll()
        {
            lock (locker)
            {
                try
                {
                    channel.QueuePurge(queue);
                    return true;
                }
                catch
                {
                    return false;
                }

            }
        }

        public void OnReceived(Action<byte[], ulong, string> action, bool autoAck = true)
        {
            try
            {
                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body.ToArray();
                    action?.Invoke(body, ea.DeliveryTag, ea.RoutingKey);
                };
                channel.BasicConsume(queue: queue,
                                     autoAck: autoAck,
                                     consumer: consumer);
            }
            catch (Exception ex)
            {
                channel.Dispose();
                throw ex;
            }

        }

        public (byte[], ulong, string) GetMessage(bool autoAck = true)
        {
            try
            {
                var res = channel.BasicGet(queue, autoAck);
                if (res == null)
                {
                    return (null, 0, null);
                }
                return (res.Body.ToArray(), res.DeliveryTag, res.RoutingKey);
            }
            catch
            {
                return (null, 0, null);
            }
        }

    }
}
